import time
import json
from confluent_kafka import KafkaError
from Slave.config import *
from confluent_kafka import Consumer
from time import sleep
from logger_config import logger_factory
from Slave.redis.redis_util import redis_change_expire_time, atomic_check_and_update
from role_config import slave_id
from Slave.config.slave_config import consumer_config, topic_name
from Slave.config.mission_config import mission_dict

logger = logger_factory()


def slave_main_consume():
    """
    实现循环不断从kafka主任务序列拉去任务并完成
    """
    # 创建kafka客户端，订阅queen_to_slave_topic
    consumer = Consumer(consumer_config)
    consumer.subscribe([topic_name])
    while True:
        try:
            # 拉取任务
            msg = consumer.poll(timeout=1.0)
            if msg is None:
                sleep(10)
                continue
            if msg.error():
                print('Kafka error: %s', msg.error())
                logger.error('Kafka error: %s', msg.error())
            else:
                # 获取消息
                msg_value = msg.value()

                # 检查消息体是否为空
                if msg_value is None or len(msg_value) == 0:
                    logger.warning('Received empty message')
                    continue

                try:
                    # 尝试解析为 JSON
                    decoded_msg = msg_value.decode('utf-8')
                    if not decoded_msg.strip():
                        logger.warning('Received blank message')
                        continue

                    # 解析任务信息，信息格式约定为json格式
                    mission_info = json.loads(decoded_msg)
                    mission_wrapper(mission_info)
                except json.JSONDecodeError as e:
                    logger.error('Failed to decode JSON: %s, message: %s', str(e), decoded_msg)
                except Exception as e:
                    print(e)
                    logger.error('Error decoding message: %s', str(e))
        except Exception as e:
            print(e)
            logger.error('Exception in main consumer: %s', str(e))
            consumer.close()
            return


def mission_wrapper(mission_info):
    mission_key = mission_info['REDIS_KEY']
    # uuid = mission_info['UUID']
    expire_time = 180
    mission_status = None
    # update_time = get_updatetime(mission_info)
    update_time = None
    if update_time is None:
        update_time = 'NaN'

    # 上锁
    ret = atomic_check_and_update(mission_key, mission_key, slave_id, 2*60*60)
    start_time = time.strftime('%Y-%m-%d %H:%M:%S',time.localtime())
    # update_start_and_update_time(uuid, mission_key, start_time, update_time)
    if ret is not True:
        logger.info('已上锁')
        return False
    logger.info('上锁成功')

    # 完成任务
    try:
        expire_time, mission_status = complete_mission(mission_info)
    except Exception as e:
        logger.info('任务执行异常: %s', str(e))
        return False
    # finally:
    #     update_status_and_finish_time(uuid, mission_key, mission_status, finish_time)
    #解锁
    ret = redis_change_expire_time(mission_key, str(expire_time))
    if ret is not True:
         logger.error('解锁失败')
         return False
    else:
        logger.info('解锁成功')
        return True


def get_update_time(mission_dict:dict):
    update_time_target_key = 'updatetime'
    up_time_target_key = 'uptime'
    for key in mission_dict:
        lower = key.lower()
        if lower == update_time_target_key or lower == up_time_target_key:
            return mission_dict[key]
    return None

def complete_mission(mission_info:dict):
    mission = mission_info['mission']
    operate = mission_dict[mission]
    expired_time, mission_status = operate(mission_info)
    return expired_time, mission_status

# def clear_kafka():
#     consumer = Consumer(slave_test_consumer_config)
#     consumer.subscribe([queen_to_slave_test_topic])
#     logger.info('start slave, ready for operate mission')
#     while True:
#         try:
#             msg = consumer.poll(timeout=1.0)
#             if msg is None:
#                 logger.info('not get mission, sleep 10 seconds')
#                 sleep(10)
#                 continue
#             if msg.error():
#                 if msg.error().code() == KafkaError.PARTITION_EOF:
#                     logger.error('%% %s [%d] reached end of offset %d\n' % (msg.topic(), msg.partition(), msg.offset()))
#             else:
#                 mission_info = json.loads(msg.value().decode('utf-8'))
#                 mission_key = mission_info['REDIS_KEY']
#                 ret = redis_change_expire_time(mission_key, str(60))
#         except Exception as e:
#             logger.error(e)
#             continue
